package com.example.demo.nio;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;

import static com.example.demo.nio.ByteBufferUtil.debugAll;

@Slf4j
public class NioServer {


    // worker负责读写事件
    static class Worker implements Runnable {
        private Thread thread;
        private Selector selector;
        private String name;
        private volatile boolean flag = false;
        private ConcurrentLinkedDeque<Runnable> queue = new ConcurrentLinkedDeque<>();

        public Worker(String name) {
            this.name = name;
        }

        // 初始化thread和selector
        public void register(SocketChannel sc) throws IOException {
            if (!flag) {
                thread = new Thread(this, name);
                selector = Selector.open();
                thread.start();
                flag = true;
            }
//            // 向队列添加任务,但该任务并没有立即执行
//            queue.add(()->{
//                try {
//                    sc.register(selector,SelectionKey.OP_READ,null);
//                } catch (ClosedChannelException e) {
//                    e.printStackTrace();
//                }
//            });
            // 唤醒select方法,避免没有事件时发生堵塞
            selector.wakeup();
            sc.register(selector,SelectionKey.OP_READ,null);

        }

        @Override
        public void run() {
            while (true) {
                // select方法没有事件发生时，线程会堵塞
                // 有事件或事件未处理select方法不会堵塞
                try {
                    // worker线程
                    selector.select();
                    Runnable task = queue.poll();
//                    if (task != null){
//                        task.run();
//                    }
                    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                    while (iter.hasNext()) {
                        SelectionKey sKey = iter.next();
                        try {
                            // 移除当前SelectionKey
                            iter.remove();
                            if (sKey.isReadable()) {
                                SocketChannel channel = (SocketChannel) sKey.channel();
                                ByteBuffer bf = ByteBuffer.allocate(16);
                                log.warn("read...{}",channel.getRemoteAddress());
//                                // 获取att
//                                ByteBuffer bf = (ByteBuffer) sKey.attachment();
                                 channel.read(bf);
//                                if (read == -1) {
//                                    sKey.cancel();
//                                } else {
//                                    split(bf);
//                                    if (bf.position() == bf.limit()) {
//                                        // bf扩容
//                                        ByteBuffer nbf = ByteBuffer.allocate(bf.capacity() * 2);
//                                        bf.flip();
//                                        // 将bf中的数据拷贝到新的bf中
//                                        nbf.put(bf);
//                                        // 设置att
//                                        sKey.attach(nbf);
//                                    }
//                                }
                                bf.flip();
                                debugAll(bf);
//                                bf.clear();
                            }
                        } catch (Exception e) {
                            sKey.cancel();
                            e.printStackTrace();
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    public static void main(String[] args) {
//        bio();
//        nio();
//        selector();
        multiSelector();
    }

    private static void multiSelector() {
        try {
            // boss负责连接事件
            Selector boss = Selector.open();
            Thread.currentThread().setName("boss");
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(8082));
            // 设置成非堵塞模式
            ssc.configureBlocking(false);
            // 一个selector可以管理多个channel
            // 将channel注册到selector
            // SelectionKey 和具体的channel绑定
            SelectionKey sscKey = ssc.register(boss, 0, null);
            // OP_ACCEPT连接事件
            sscKey.interestOps(SelectionKey.OP_ACCEPT);

            // 获取线程的核心数
            Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];
            for (int i = 0; i < workers.length; i++) {
                workers[i] = new Worker("worker-" + i);
            }
            AtomicInteger i = new AtomicInteger();
            while (true){
                // select方法没有事件发生时，线程会堵塞
                // 有事件或事件未处理select方法不会堵塞
                boss.select();
                // 处理事件
                // 获取所有事件
                Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
                while (iter.hasNext()){
                    SelectionKey sKey = iter.next();
                    try {
                        // 移除当前SelectionKey
                        iter.remove();
                        // 连接事件
                        if (sKey.isAcceptable()) {
                            ServerSocketChannel channel = (ServerSocketChannel) sKey.channel();
                            SocketChannel sc = channel.accept();
                            sc.configureBlocking(false);
                            log.warn("connected...{}",sc.getRemoteAddress());
                            log.warn("before register...{}",sc.getRemoteAddress());
                            // main线程关联worker
                            workers[i.getAndIncrement() % workers.length].register(sc);
//                            sc.register(worker0.selector,SelectionKey.OP_READ,null);
                            log.warn("after register...{}",sc.getRemoteAddress());
                        }
                    } catch (Exception e) {
                        sKey.cancel();
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();

        }
    }

    private static void split(ByteBuffer source) {
        source.flip();
        for (int i = 0; i < source.limit(); i++) {
            if (source.get(i) == '\n'){
                int length = i + 1 - source.position();
                ByteBuffer target = ByteBuffer.allocate(length);
                for (int j = 0; j < length; j++) {
                    target.put(source.get());
                }
                debugAll(target);
            }
        }
        source.compact();
    }

    private static void selector() {
        try {
            // 一个selector可以管理多个channel
            Selector selector = Selector.open();
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(8082));
            // 设置成非堵塞模式
            ssc.configureBlocking(false);
            // 将channel注册到selector
            // SelectionKey 和具体的channel绑定
            SelectionKey sscKey = ssc.register(selector, 0, null);
            // OP_ACCEPT连接事件
            sscKey.interestOps(SelectionKey.OP_ACCEPT);
            while (true){
                // select方法没有事件发生时，线程会堵塞
                // 有事件或事件未处理select方法不会堵塞
                selector.select();
                // 处理事件
                // 获取所有事件
                Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                while (iter.hasNext()){
                    SelectionKey sKey = iter.next();
                    try {
                        // 移除当前SelectionKey
                        iter.remove();
                        // 连接事件
                        if (sKey.isAcceptable()) {
                            ServerSocketChannel channel = (ServerSocketChannel) sKey.channel();
                            SocketChannel sc = channel.accept();
                            sc.configureBlocking(false);
                            ByteBuffer bf = ByteBuffer.allocate(16);
                            // 将bf作为attachment跟着channel注册到selector
                            SelectionKey scKey = sc.register(selector, 0, bf);
                            // OP_READ读事件
                            scKey.interestOps(SelectionKey.OP_READ);
                        // 读取事件
                        }else if (sKey.isReadable()){
                            SocketChannel channel = (SocketChannel) sKey.channel();
                            // 获取att
                            ByteBuffer bf = (ByteBuffer) sKey.attachment();
                            int read = channel.read(bf);
                            if (read == -1){
                                sKey.cancel();
                            }else {
                                split(bf);
                                if (bf.position() == bf.limit()){
                                    // bf扩容
                                    ByteBuffer nbf = ByteBuffer.allocate(bf.capacity() * 2);
                                    bf.flip();
                                    // 将bf中的数据拷贝到新的bf中
                                    nbf.put(bf);
                                    // 设置att
                                    sKey.attach(nbf);
                                }
                            }
//                            bf.flip();
//                            debugAll(bf);
//                            bf.clear();
                        }
                    } catch (Exception e) {
                        sKey.cancel();
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();

        }
    }

    private static void nio() {
        try {
            ByteBuffer bf = ByteBuffer.allocate(16);
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(8081));
            ssc.configureBlocking(false);
            List<SocketChannel> channels = new ArrayList<>();
            while (true){

                SocketChannel sc = ssc.accept();
                if (sc != null){
                    log.debug("------连接中-------");
                    sc.configureBlocking(false);
                    channels.add(sc);
                    log.debug("------连接成功-------");
                }
                for (SocketChannel channel : channels){
                    int read = channel.read(bf);
                    if (read > 0){
                        log.debug("------读取buffer-------",channel);
                        bf.flip();
                        debugAll(bf);
                        bf.clear();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    private static void bio() {
        try {
            ByteBuffer bf = ByteBuffer.allocate(16);
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(8081));
            List<SocketChannel> channels = new ArrayList<>();
            while (true){
                log.debug("------连接中-------");
                // 建立连接
                SocketChannel sc = ssc.accept();
                channels.add(sc);
                log.debug("------连接成功-------");
                for (SocketChannel channel : channels){
                    log.debug("------读取buffer前-------",channel);
                    // 读取buffer
                    channel.read(bf);
                    log.debug("------读取buffer后-------",channel);
                    // 切换读模式
                    bf.flip();
                    debugAll(bf);
                    // 清除buffer
                    bf.clear();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
